package ExperimentSix.Tensorflow;

import java.io.*;
import java.net.*;
import java.util.*;
import java.util.concurrent.*;

public class Master implements Runnable {
    private ServerSocket serverSocket;
    private int totalWorkersPerMaster;
    private int taskCount;
    private List<Integer> results = Collections.synchronizedList(new ArrayList<>());
    private int masterId;
    private static final int TOTAL_MASTERS = 2; // 总Master数量

    public Master(int port, int masterId, int totalWorkersPerMaster, int taskCount) throws IOException {
        this.serverSocket = new ServerSocket(port);
        this.masterId = masterId;
        this.totalWorkersPerMaster = totalWorkersPerMaster;
        this.taskCount = taskCount;
    }

    @Override
    public void run() {
        ExecutorService executor = Executors.newFixedThreadPool(totalWorkersPerMaster);
        try {
            System.out.println("Master " + masterId + " 运行在端口" + serverSocket.getLocalPort());
            for (int i = 0; i < totalWorkersPerMaster; i++) {
                Socket workerSocket = serverSocket.accept();
                executor.submit(new TaskDistributor(workerSocket, taskCount, this));
            }
            executor.shutdown();

            // 添加等待所有任务完成的逻辑，包括超时处理
            try {
                if (!executor.awaitTermination(1, TimeUnit.MINUTES)) { // 等待一分钟，可根据实际情况调整
                    System.err.println("Master " + masterId + ": Timed out waiting for tasks to finish, shutting down now.");
                    executor.shutdownNow(); // 强制关闭，取消未完成的任务
                }
            } catch (InterruptedException e) {
                executor.shutdownNow(); // 如果当前线程被中断，也应取消任务
                Thread.currentThread().interrupt(); // 重新设置中断状态
            }

            System.out.println("Master " + masterId + ": 所有任务完成.");
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                serverSocket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
//master收集统计结果
    synchronized void addResult(int result) {
        results.add(result);
        if (results.size() == totalWorkersPerMaster * taskCount) {
            System.out.println("Master " + masterId + ": 接收完所有结果. Results: " + results);
            // 结果统计求平均值逻辑
            int sum = 0;
            for (int res : results) {
                sum += res;
            }
            double average = (double) sum / results.size();
            System.out.println("Master " + masterId + ": Summary - Sum: " + sum + ", Average: " + average);
            // 进一步处理或与其他Master同步结果的逻辑
            System.out.println("Master " + masterId + ": 发送其他任务");

        }
    }

    public static void main(String[] args) {
        try {
            // 启动两个Master节点，使用不同的端口
            new Thread(new Master(55952, 1, 2, 5)).start(); // Master 1管理前2个Worker
            new Thread(new Master(55953, 2, 1, 4)).start(); // Master 2管理剩下的1个Worker
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}